package org.apache.flink.runtime.resourceconsumer.factories;

import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import org.apache.elasticresource.api.BusinessResourceConsumer;
import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.configuration.ConsumerManagerOptions;
import org.apache.flink.runtime.allocatedeploy.DefaultExecutionDeployer;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.consumermanager.OnCompletionActions;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.resourceconsumer.ConsumingMasterId;
import org.apache.flink.runtime.resourceconsumer.DefaultExecutionDeploymentReconciler;
import org.apache.flink.runtime.resourceconsumer.DefaultExecutionDeploymentTracker;
import org.apache.flink.runtime.resourceconsumer.ResourceConsumer;
import org.apache.flink.runtime.resourceconsumer.ResourceConsumerConfiguration;
import org.apache.flink.runtime.resourceconsumer.ResourceConsumerService;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.slotpool.SlotPoolServiceFactory;
import org.apache.flink.runtime.util.Hardware;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.apache.flink.util.function.FunctionUtils;

/**
 * @类名: DefaultResourceConsumerServiceFactory
 * @说明: ResourceConsumer工厂默认实现，ResourceConsumer是ResourceConsumerService实现
 *
 * @author   leehom
 * @Date	 2023年11月1日 下午3:41:26
 * 修改记录：
 *
 * @see 	 
 */
public class DefaultResourceConsumerServiceFactory implements ResourceConsumerServiceFactory {

	private final ConsumerID consumerId;
	private final String consumerName;
    private final Executor executor;
    private final RpcService rpcService;
    private final ResourceConsumerConfiguration resourceConsumerConfiguration;
    private final HighAvailabilityServices haServices;
    private final SlotPoolServiceFactory slotPoolServiceFactory;
    private final HeartbeatServices heartbeatServices;
    private final ConsumerManagerConsumerMetricGroupFactory consumerManagerConsumerMetricGroupFactory;
    private final FatalErrorHandler fatalErrorHandler;
    private final BusinessResourceConsumer businessResourceConsumer;

    public DefaultResourceConsumerServiceFactory(
    		ConsumerID consumerId,
    		String consumerName,
            Executor executor,
            RpcService rpcService,
            ResourceConsumerConfiguration resourceConsumerConfiguration,
            HighAvailabilityServices haServices,
            SlotPoolServiceFactory slotPoolServiceFactory,
            HeartbeatServices heartbeatServices,
            ConsumerManagerConsumerMetricGroupFactory consumerManagerConsumerMetricGroupFactory,
            FatalErrorHandler fatalErrorHandler,
            long initializationTimestamp,
            BusinessResourceConsumer businessResourceConsumer) {
    	this.consumerId = consumerId;
    	this.consumerName = consumerName;
        this.executor = executor;
        this.rpcService = rpcService;
        this.resourceConsumerConfiguration = resourceConsumerConfiguration;
        this.haServices = haServices;
        this.slotPoolServiceFactory = slotPoolServiceFactory;
        this.heartbeatServices = heartbeatServices;
        this.consumerManagerConsumerMetricGroupFactory = consumerManagerConsumerMetricGroupFactory;
        this.fatalErrorHandler = fatalErrorHandler;
        this.businessResourceConsumer = businessResourceConsumer;
    }

    // create&start ResourceConsumer
    @Override
    public CompletableFuture<ResourceConsumerService> createResourceConsumerService(
            UUID leaderSessionId, OnCompletionActions onCompletionActions) {
        return CompletableFuture.supplyAsync(
                FunctionUtils.uncheckedSupplier(
                        () -> internalCreateResourceConsumerService(leaderSessionId, onCompletionActions)),
                executor);
    }

    private ResourceConsumer internalCreateResourceConsumerService(
            UUID leaderSessionId, OnCompletionActions onCompletionActions) throws Exception {
        final int numberCPUCores = Hardware.getNumberCPUCores();
        final int jobManagerFuturePoolSize =
        		resourceConsumerConfiguration.getConfiguration().getInteger(ConsumerManagerOptions.JOB_MANAGER_FUTURE_POOL_SIZE, numberCPUCores);
        final ScheduledExecutorService futureExecutor =
                Executors.newScheduledThreadPool(
                        jobManagerFuturePoolSize, new ExecutorThreadFactory("consumermanager-future"));
        //
        final ResourceConsumer resourceConsumer =
                new ResourceConsumer(
                        ConsumingMasterId.fromUuidOrNull(leaderSessionId),
                		consumerName,
                        rpcService,
                        resourceConsumerConfiguration,
                        ResourceID.generate(),
                        consumerId, 
                        haServices,
                        slotPoolServiceFactory,
                        heartbeatServices,
                        consumerManagerConsumerMetricGroupFactory,
                        onCompletionActions,
                        futureExecutor, 
                        futureExecutor, 
                        fatalErrorHandler,
                        new DefaultExecutionDeployer.Factory(),
                        businessResourceConsumer,
                        new DefaultExecutionDeploymentTracker(),
                        DefaultExecutionDeploymentReconciler::new);
        // 启动
        resourceConsumer.start();
        //
        return resourceConsumer;
    }
    
}
